package org.apache.hadoop.hdfs.server.datanode;

import static org.apache.hadoop.util.Time.now;

import java.io.EOFException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import com.google.common.base.Joiner;
import org.apache.commons.logging.Log;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
import org.apache.hadoop.hdfs.client.BlockReportOptions;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.StorageType;
import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.RollingUpgradeStatus;
import org.apache.hadoop.hdfs.protocol.UnregisteredNodeException;
import org.apache.hadoop.hdfs.protocolPB.DatanodeProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdfs.server.common.IncorrectVersionException;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.hdfs.server.protocol.BlockReportContext;
import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand;
import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
import org.apache.hadoop.hdfs.server.protocol.DisallowedDatanodeException;
import org.apache.hadoop.hdfs.server.protocol.HeartbeatResponse;
import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
import org.apache.hadoop.hdfs.server.protocol.ReceivedDeletedBlockInfo;
import org.apache.hadoop.hdfs.server.protocol.StorageBlockReport;
import org.apache.hadoop.hdfs.server.protocol.StorageReceivedDeletedBlocks;
import org.apache.hadoop.hdfs.server.protocol.StorageReport;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.util.VersionInfo;
import org.apache.hadoop.util.VersionUtil;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;

/**
 * 每个活跃active或备份standby状态NameNode对应的线程，它负责完成以下操作：
 * 1、与NameNode进行预登记握手；
 * 2、在NameNode上注册；
 * 3、发送周期性的心跳给NameNode；
 * 4、处理从NameNode接收到的请求。
 */
@InterfaceAudience.Private
// 实现Runnable接口意味着BPServiceActor是一个线程
class BPServiceActor implements Runnable {
  
  static final Log LOG = DataNode.LOG;
  /** NameNode地址 */
  final InetSocketAddress nnAddr;
  /** HA服务状态 */
  HAServiceState state;
  /** BPServiceActor线程所属BPOfferService */
  final BPOfferService bpos;
  
  volatile long lastBlockReport = 0;
  volatile long lastDeletedReport = 0;

  boolean resetBlockReportTime = true;

  volatile long lastCacheReport = 0;

  Thread bpThread;
  /**
   * NameNode在DataNode上的代理bpNamenode，
   * 它是BPServiceActor线程中一个DatanodeProtocolClientSideTranslatorPB类型的变量，
   * 也就意味着每个与NameNode通讯的BPServiceActor工作线程，都持有一个NameNode的代理，其初始化是在BPServiceActor工作线程与NameNode连接时完成的
   */
  // 这个东西是一个rpc调用的代理，代表了namenode, 通过这个代表了namenode的rpc通信，可以让datanode跟namenode进行rpc的通信及调用
  DatanodeProtocolClientSideTranslatorPB bpNamenode;
  private volatile long lastHeartbeat = 0;

  /**
   * CONNECTING 正在连接
   * INIT_FAILED 初始化失败
   * RUNNING 正在运行
   * EXITED 已退出
   * FAILED 已失败
   */
  static enum RunningState {
    CONNECTING, INIT_FAILED, RUNNING, EXITED, FAILED;
  }

  /** 运行状态runningState默认为枚举类RunningState的CONNECTING，表示正在连接 */
  private volatile RunningState runningState = RunningState.CONNECTING;

  /**
   * 在数据块汇报（通常一小时一次）之间，DataNode会汇报其数据块列表的增量变化情况。
   * 这个Map，包含尚未汇报给NameNode的DataNode上数据块正在发生的变化。
   * 访问它必须使用synchronized关键字。
   *
   * pendingIncrementalBRperStorage变量对应的数据结构，
   * 它是一个Map，key为DatanodeStorage类型，value为PerStoragePendingIncrementalBR类型。
   * 而这个PerStoragePendingIncrementalBR类型在其内部封装了一个叫做pendingIncrementalBR的HashMap， key为blockId，value为ReceivedDeletedBlockInfo，
   * ReceivedDeletedBlockInfo对Block做了一层封装了，它标识了对应Block在DataNode上的状态BlockStatus，
   * BlockStatus是一个枚举类，包含的Block状态分别有正在接收的数据块RECEIVING_BLOCK(1)、已经接收的数据块RECEIVED_BLOCK(2)、已被删除的数据块DELETED_BLOCK(3)三种状态。
   *
   * pendingIncrementalBRperStorage实际上存储了DataNode上每个DatanodeStorage到对应的增量数据块集合的映射关系，而这个增量数据块，包含正在接收的、已接受的和已删除的。
   *
   * 在数据块汇报（通常一小时一次）之间，DataNode会汇报其数据块列表的增量变化情况，这个是作为一个小的（smaller）汇报进行的。
   * 这个Map，包含尚未汇报给NameNode的DataNode上数据块正在发生的变化，访问它必须使用synchronized关键字。
   * 而这个数据块增量汇报，其主要目的就应该是尽早让NameNode了解数据节点DataNode上数据块的变化情况，而不是通过正常的每小时一次的数据块汇报来告知名字节点， 那样的话对于整个文件系统来说，是很被动的一见事。
   */
  private final Map<DatanodeStorage, PerStoragePendingIncrementalBR> pendingIncrementalBRperStorage = Maps.newHashMap();

  // IBR = Incremental Block Report. If this flag is set then an IBR will be
  // sent immediately by the actor thread without waiting for the IBR timer to elapse.
  private volatile boolean sendImmediateIBR = false;
  private volatile boolean shouldServiceRun = true;
  private final DataNode dn;
  private final DNConf dnConf;

  private DatanodeRegistration bpRegistration;

  // 构造方法，BPServiceActor被创建时就已明确知道NameNode地址InetSocketAddress类型的nnAddr，和BPOfferService类型的bpos
  BPServiceActor(InetSocketAddress nnAddr, BPOfferService bpos) {
    this.bpos = bpos;
    this.dn = bpos.getDataNode();
    this.nnAddr = nnAddr;
    this.dnConf = dn.getDnConf();
  }

  boolean isAlive() {
    if (!shouldServiceRun || !bpThread.isAlive()) {
      return false;
    }
    return runningState == BPServiceActor.RunningState.RUNNING
        || runningState == BPServiceActor.RunningState.CONNECTING;
  }

  @Override
  public String toString() {
    return bpos.toString() + " service to " + nnAddr;
  }
  
  InetSocketAddress getNNSocketAddress() {
    return nnAddr;
  }

  /**
   * Used to inject a spy NN in the unit tests.
   */
  @VisibleForTesting
  void setNameNode(DatanodeProtocolClientSideTranslatorPB dnProtocol) {
    bpNamenode = dnProtocol;
  }

  @VisibleForTesting
  DatanodeProtocolClientSideTranslatorPB getNameNodeProxy() {
    return bpNamenode;
  }

  /**
   * Perform the first part of the handshake with the NameNode.
   * This calls <code>versionRequest</code> to determine the NN's
   * namespace and version info. It automatically retries until
   * the NN responds or the DN is shutting down.
   * 
   * @return the NamespaceInfo
   */
  @VisibleForTesting
  NamespaceInfo retrieveNamespaceInfo() throws IOException {
    NamespaceInfo nsInfo = null;
    while (shouldRun()) {
      try {
        // 相当于在底层bpNamenode作为rpc代理，
        // bpNamenode底层会构造一个网络请求，去连接到namenode的rpc server端口上去，发送一个请求， 这个请求说明了要调用versionRequest()方法，
        // rpc server会统一处理所有的rpc调用的网络连接和请求，
        // rpc server会将对应的接口方法的请求，转发给我们看到的哪些小service具体里处理某个接口的请求
        // rpc server最后会将响应结果返回给datanode
        nsInfo = bpNamenode.versionRequest();
        LOG.debug(this + " received versionRequest response: " + nsInfo);
        break;
      } catch(SocketTimeoutException e) {
        LOG.warn("Problem connecting to server: " + nnAddr);
      } catch(IOException e ) {
        LOG.warn("Problem connecting to server: " + nnAddr);
      }
      // try again in a second
      sleepAndLogInterrupts(5000, "requesting version info from NN");
    }
    
    if (nsInfo != null) {
      // 检查软件版本最小的版本号，检查datanode的版本号和namenode的版本后是否一致
      checkNNVersion(nsInfo);
    } else {
      throw new IOException("DN shut down before block pool connected");
    }
    return nsInfo;
  }

  private void checkNNVersion(NamespaceInfo nsInfo) throws IncorrectVersionException {
    // build and layout versions should match
    String nnVersion = nsInfo.getSoftwareVersion();
    String minimumNameNodeVersion = dnConf.getMinimumNameNodeVersion();
    if (VersionUtil.compareVersions(nnVersion, minimumNameNodeVersion) < 0) {
      IncorrectVersionException ive = new IncorrectVersionException( minimumNameNodeVersion, nnVersion, "NameNode", "DataNode");
      LOG.warn(ive.getMessage());
      throw ive;
    }

    String dnVersion = VersionInfo.getVersion();
    if (!nnVersion.equals(dnVersion)) {
      LOG.info("Reported NameNode version '" + nnVersion + "' does not match DataNode version '" + dnVersion + "' but is within acceptable limits. Note: This is normal during a rolling upgrade.");
    }
  }

  /**
   * 连接NameNode并握手， 它的主要处理流程如下：
   *   1、利用DataNode实例dn的connectToNN()方法和NameNode地址nnAddr获得NameNode的代理 bpNamenode
   *   2、与NameNode握手第1阶段：调用retrieveNamespaceInfo()方法获取命名空间信息nsInfo；
   *   3、通过bpos的verifyAndSetNamespaceInfo()方法进行验证，并设置命名空间信息nsInfo；
   *   4、与NameNode握手第2阶段，调用register()方法进行注册。
   */
  private void connectToNNAndHandshake() throws IOException {
    // 利用DataNode实例dn的connectToNN()方法和NameNode地址nnAddr获得NameNode的代理bpNamenode
    // 连接到namenode上去，获取一个 bpNamenode
    // bpNamenode就是一个负责进行rpc接口调用的一个东西，通过这个东西就可以让BPServiceActor以rpc接口调用的方式，去请求namenode
    // bpNamenode就是一个rpc接口调用的代理，在datanode获取了一个rpc代理之后，后面就可以通过这个代理跟namenode进行通信
    bpNamenode = dn.connectToNN(nnAddr);

    // 与namenode握手的第1阶段： 获取namespace信息.
    // 通过上面的那个bpNamenode作为一个rpc代理，调用了namenode的一个rpc接口，获取到了一个namenode的namespaceInfo
    // NamespaceInfo 包含了一些id，比如clusterId, nsId,blockPoolId, namenode的id

    // datanode通过rpc接口的调用，调用namenode的接口
    NamespaceInfo nsInfo = retrieveNamespaceInfo(); // TODO

    // 第1件事情：两个namenode都会返回一个NamespaceInfo，此时就会在第2个namenode返回NamespaceInfo的时候，进行两个namespaceInfo的校验，校验一下他们两个其实是一样的。
    // 第2件事情：如果第1个namenode返回了一个NamespaceInfo之后，一定会在dtanode这里初始化DataStorage，里面会初始胡对应的block存储空进，启动一些block相关的后台线程

    // 验证，并设置命名空间信息
    bpos.verifyAndSetNamespaceInfo(nsInfo); // TODO 核心代码
    
    // 与namenode握手的第2阶段。注册
    register(); // TODO 核心代码
  }

  // This is useful to make sure NN gets Heartbeat before Blockreport
  // upon NN restart while DN keeps retrying Otherwise,
  // 1. NN restarts.
  // 2. Heartbeat RPC will retry and succeed. NN asks DN to reregister.
  // 3. After reregistration completes, DN will send Blockreport first.
  // 4. Given NN receives Blockreport after Heartbeat, it won't mark
  //    DatanodeStorageInfo#blockContentsStale to false until the next
  //    Blockreport.
  void scheduleHeartbeat() {
    lastHeartbeat = 0;
  }

  /**
   * This methods  arranges for the data node to send the block report at 
   * the next heartbeat.
   */
  void scheduleBlockReport(long delay) {
    if (delay > 0) { // send BR after random delay
      lastBlockReport = Time.now() - ( dnConf.blockReportInterval - DFSUtil.getRandom().nextInt((int)(delay)));
    } else { // send at next heartbeat
      lastBlockReport = lastHeartbeat - dnConf.blockReportInterval;
    }
    resetBlockReportTime = true; // reset future BRs for randomness
  }

  void reportBadBlocks(ExtendedBlock block,
      String storageUuid, StorageType storageType) {
    if (bpRegistration == null) {
      return;
    }
    DatanodeInfo[] dnArr = { new DatanodeInfo(bpRegistration) };
    String[] uuids = { storageUuid };
    StorageType[] types = { storageType };
    LocatedBlock[] blocks = { new LocatedBlock(block, dnArr, uuids, types) };
    
    try {
      bpNamenode.reportBadBlocks(blocks);  
    } catch (IOException e){
      /* One common reason is that NameNode could be in safe mode.
       * Should we keep on retrying in that case?
       */
      LOG.warn("Failed to report bad block " + block + " to namenode : "
          + " Exception", e);
    }
  }
  
  /**
   *大致处理流程如下：
   * 1、创建一个存储StorageReceivedDeletedBlocks的ArrayList列表reports：
   *      大小为pendingIncrementalBRperStorage的大小。
   *      StorageReceivedDeletedBlocks是对DatanodeStorage和ReceivedDeletedBlockInfo数组的一个封装，
   *      实际上就是将pendingIncrementalBRperStorage由Map转换为List列表形式；
   * 2、使用synchronized对pendingIncrementalBRperStorage进行同步，遍历pendingIncrementalBRperStorage：
   *      2.1、取出每个DatanodeStorage、PerStoragePendingIncrementalBR进行处理；
   *      2.2、如果perStorageMap中存在发生变化的数据块，发送新接收的或者已删除的数据块ID给NameNode：
   *          2.2.1、从perStorageMap中获得ReceivedDeletedBlockInfo数组；
   *          2.2.2、将根据DatanodeStorage和ReceivedDeletedBlockInfo数组构造的StorageReceivedDeletedBlocks加入reports列表；
   * 3、立即汇报的标志位sendImmediateIBR设置为false；
   * 4、reports大小为0的话，直接返回null；
   * 5、发送是否成功的标志位success初始化为false；
   * 6、通过NameNode代理bpNamenode的blockReceivedAndDeleted()方法，将新接收的或者已删除的数据块汇报给NameNode，汇报的信息包括：
   *      6.1、数据节点注册信息DatanodeRegistration；
   *      6.2、数据块池ID；
   *      6.3、需要汇报的数据块及其状态信息列表StorageReceivedDeletedBlocks；
   * 7、发送是否成功的标志位success设置为true；
   * 8、汇报不成功的话，遍历reports：
   *      8.1、将数据块再放回到perStorageMap；
   *      8.2、立即汇报的标志位sendImmediateIBR设置为true。
   */
  private void reportReceivedDeletedBlocks() throws IOException {

    // 创建一个存储StorageReceivedDeletedBlocks的ArrayList列表reports，
    // 大小为pendingIncrementalBRperStorage的大小
    // StorageReceivedDeletedBlocks是对DatanodeStorage和ReceivedDeletedBlockInfo数组的一个封装，
    // 实际上就是将pendingIncrementalBRperStorage由Map转换为List列表形式
    ArrayList<StorageReceivedDeletedBlocks> reports = new ArrayList<StorageReceivedDeletedBlocks>(pendingIncrementalBRperStorage.size());

    // 使用synchronized对pendingIncrementalBRperStorage进行同步：
    synchronized (pendingIncrementalBRperStorage) {
      // 遍历pendingIncrementalBRperStorage
      for (Map.Entry<DatanodeStorage, PerStoragePendingIncrementalBR> entry : pendingIncrementalBRperStorage.entrySet()) {
        // 取出每个DatanodeStorage、PerStoragePendingIncrementalBR进行处理
        final DatanodeStorage storage = entry.getKey();
        final PerStoragePendingIncrementalBR perStorageMap = entry.getValue();

        // 如果perStorageMap中存在发生变化的数据块：
        if (perStorageMap.getBlockInfoCount() > 0) {
          // 发送新接收的或者已删除的数据块ID给NameNode

          // 从perStorageMap中获得ReceivedDeletedBlockInfo数组
          ReceivedDeletedBlockInfo[] rdbi = perStorageMap.dequeueBlockInfos();
          // 将根据DatanodeStorage和ReceivedDeletedBlockInfo数组构造的StorageReceivedDeletedBlocks加入reports列表
          reports.add(new StorageReceivedDeletedBlocks(storage, rdbi));
        }
      }

      // 立即汇报的标志位sendImmediateIBR设置为false
      sendImmediateIBR = false;
    }

    if (reports.size() == 0) { // reports大小为0的话，直接返回null
      return;
    }

    // 发送是否成功的标志位success初始化为false
    boolean success = false;
    try {
      // 通过NameNode代理的blockReceivedAndDeleted()方法，将新接收的或者已删除的数据块汇报给NameNode，汇报的信息包括：
      // 1、数据节点注册信息DatanodeRegistration；
      // 2、数据块池ID；
      // 3、需要汇报的数据块及其状态信息列表StorageReceivedDeletedBlocks；
      bpNamenode.blockReceivedAndDeleted(bpRegistration, bpos.getBlockPoolId(), reports.toArray(new StorageReceivedDeletedBlocks[reports.size()]));
      // 发送是否成功的标志位success设置为true
      success = true;
    } finally {
      if (!success) { // 汇报不成功的话
        synchronized (pendingIncrementalBRperStorage) {
          for (StorageReceivedDeletedBlocks report : reports) {
            // 将数据块再放回到perStorageMap
            PerStoragePendingIncrementalBR perStorageMap = pendingIncrementalBRperStorage.get(report.getStorage());
            perStorageMap.putMissingBlockInfos(report.getBlocks());
            // 立即汇报的标志位sendImmediateIBR设置为true
            sendImmediateIBR = true;
          }
        }
      }
    }
  }

  /**
   * @return pending incremental block report for given {@code storage}
   */
  private PerStoragePendingIncrementalBR getIncrementalBRMapForStorage(
      DatanodeStorage storage) {
    PerStoragePendingIncrementalBR mapForStorage =
        pendingIncrementalBRperStorage.get(storage);

    if (mapForStorage == null) {
      // This is the first time we are adding incremental BR state for
      // this storage so create a new map. This is required once per
      // storage, per service actor.
      mapForStorage = new PerStoragePendingIncrementalBR();
      pendingIncrementalBRperStorage.put(storage, mapForStorage);
    }

    return mapForStorage;
  }

  /**
   * Add a blockInfo for notification to NameNode. If another entry
   * exists for the same block it is removed.
   *
   * Caller must synchronize access using pendingIncrementalBRperStorage.
   */
  void addPendingReplicationBlockInfo(ReceivedDeletedBlockInfo bInfo,
      DatanodeStorage storage) {
    // Make sure another entry for the same block is first removed.
    // There may only be one such entry.
    for (Map.Entry<DatanodeStorage, PerStoragePendingIncrementalBR> entry :
          pendingIncrementalBRperStorage.entrySet()) {
      if (entry.getValue().removeBlockInfo(bInfo)) {
        break;
      }
    }
    getIncrementalBRMapForStorage(storage).putBlockInfo(bInfo);
  }

  /**
   * 通知namenode可能需要很长时间!
   * 我们是否应该等到namenode收到通知后，才能成功响应客户机？现在我们没有。
   */
  void notifyNamenodeBlock(ReceivedDeletedBlockInfo bInfo, String storageUuid, boolean now) {
    synchronized (pendingIncrementalBRperStorage) {
      addPendingReplicationBlockInfo( bInfo, dn.getFSDataset().getStorage(storageUuid));
      sendImmediateIBR = true;
      // 如果 now=true，报告马上就发出去。 否则，它将在下一次心跳时发出
      if (now) {
        pendingIncrementalBRperStorage.notifyAll();
      }
    }
  }

  void notifyNamenodeDeletedBlock(ReceivedDeletedBlockInfo bInfo, String storageUuid) {
    synchronized (pendingIncrementalBRperStorage) {
      addPendingReplicationBlockInfo(bInfo, dn.getFSDataset().getStorage(storageUuid));
    }
  }

  /**
   * Run an immediate block report on this thread. Used by tests.
   */
  @VisibleForTesting
  void triggerBlockReportForTests() {
    synchronized (pendingIncrementalBRperStorage) {
      lastBlockReport = 0;
      lastHeartbeat = 0;
      pendingIncrementalBRperStorage.notifyAll();
      while (lastBlockReport == 0) {
        try {
          pendingIncrementalBRperStorage.wait(100);
        } catch (InterruptedException e) {
          return;
        }
      }
    }
  }
  
  @VisibleForTesting
  void triggerHeartbeatForTests() {
    synchronized (pendingIncrementalBRperStorage) {
      lastHeartbeat = 0;
      pendingIncrementalBRperStorage.notifyAll();
      while (lastHeartbeat == 0) {
        try {
          pendingIncrementalBRperStorage.wait(100);
        } catch (InterruptedException e) {
          return;
        }
      }
    }
  }

  @VisibleForTesting
  void triggerDeletionReportForTests() {
    synchronized (pendingIncrementalBRperStorage) {
      lastDeletedReport = 0;
      pendingIncrementalBRperStorage.notifyAll();

      while (lastDeletedReport == 0) {
        try {
          pendingIncrementalBRperStorage.wait(100);
        } catch (InterruptedException e) {
          return;
        }
      }
    }
  }

  @VisibleForTesting
  boolean hasPendingIBR() {
    return sendImmediateIBR;
  }

  private long prevBlockReportId = 0;

  private long generateUniqueBlockReportId() {
    long id = System.nanoTime();
    if (id <= prevBlockReportId) {
      id = prevBlockReportId + 1;
    }
    prevBlockReportId = id;
    return id;
  }

  /**
   * 向Namenode报告blocks列表 处理流程大体如下：
   * 1、取当前开始时间startTime；
   * 2、如果当前时间startTime减去上次数据块汇报时间小于数据节点配置的数据块汇报时间间隔的话，直接返回null：
   *      数据节点配置的数据块汇报时间间隔取参数dfs.blockreport.intervalMsec，参数未配置的话默认为6小时；
   * 3、构造数据节点命令ArrayList列表cmds，存储数据块汇报返回的命令DatanodeCommand；
   * 4、调用reportReceivedDeletedBlocks()方法发送数据块增量汇报；
   * 5、记录上次数据块增量汇报时间lastDeletedReport；
   * 6、设置数据块汇报起始时间brCreateStartTime为当前时间；
   * 7、从数据节点DataNode根据线程对应块池ID获取数据块汇报集合perVolumeBlockLists：
   *      key为数据节点存储DatanodeStorage，value为数据节点存储所包含的Long类数据块数组BlockListAsLongs；
   * 8、创建数据块汇报数组StorageBlockReport，大小为上述perVolumeBlockLists的大小；
   * 9、取出value：BlockListAsLongs：
   *      9.1、取出value：BlockListAsLongs；
   *      9.2、将BlockListAsLongs封装成StorageBlockReport加入数据块汇报数组reports，StorageBlockReport包含数据节点存储DatanodeStorage和其上数据块数组；
   *      9.3、累加数据块数目totalBlockCount;
   * 10、根据数据块总数目判断是否需要多次发送消息：
   *      10.1、如果数据块总数目在split阈值之下，则将所有的数据块汇报信息放在一个消息中发送（split阈值取参数dfs.blockreport.split.threshold，参数未配置的话默认为1000*1000）：
   *          10.1.1、发送的数据块汇报消息数numReportsSent设置为1；
   *          10.1.2、通过NameNode代理bpNamenode的blockReport()方法向NameNode发送数据块汇报信息；
   *          10.1.3、将数据块汇报后返回的命令cmd加入到命令列表cmds；
   *      10.2、如果数据块总数目在split阈值之上，将数据块汇报按照DatanodeStorage分多个消息来发送：
   *          10.2.1、发送的数据块汇报消息数numReportsSent设置为i，即DatanodeStorage数目；
   *          10.2.2、遍历reports，取出每个StorageBlockReport：
   *              10.2.2.1、通过NameNode代理bpNamenode的blockReport()方法向NameNode发送数据块汇报信息；
   *              10.2.2.2、将数据块汇报后返回的命令cmd加入到命令列表cmds；
   * 11 、计算数据块汇报耗时并记录在日志Log、数据节点Metrics指标体系中；
   * 12、调用scheduleNextBlockReport()方法，调度下一次数据块汇报；
   * 13、返回命令cmds。
   */
  List<DatanodeCommand> blockReport() throws IOException {
    // 到期就发送数据块汇报

    // 取当前开始时间startTime
    final long startTime = now();
    // 这边是在汇报block，当前时间 - 上一次block report的时间 > 间隔时间（默认6小时），就可以发送一次block report
    if (startTime - lastBlockReport <= dnConf.blockReportInterval) {
      return null;
    }

    // 构造数据节点命令ArrayList列表cmds，存储数据块汇报返回的命令DatanodeCommand
    final ArrayList<DatanodeCommand> cmds = new ArrayList<DatanodeCommand>();

    // 调用reportReceivedDeletedBlocks()方法发送数据块增量汇报
    reportReceivedDeletedBlocks(); //
    // 记录上次数据块增量汇报时间lastDeletedReport
    lastDeletedReport = startTime;
    // 设置数据块汇报起始时间brCreateStartTime为当前时间
    long brCreateStartTime = now();
    // 从数据节点DataNode根据线程对应块池ID获取数据块汇报集合perVolumeBlockLists，
    // key为数据节点存储DatanodeStorage，value为数据节点存储所包含的Long类数据块数组BlockListAsLongs
    Map<DatanodeStorage, BlockListAsLongs> perVolumeBlockLists = dn.getFSDataset().getBlockReports(bpos.getBlockPoolId());

    // Convert the reports to the format expected by the NN.
    int i = 0;
    int totalBlockCount = 0;

    // 创建数据块汇报数组StorageBlockReport，大小为上述perVolumeBlockLists的大小
    StorageBlockReport reports[] = new StorageBlockReport[perVolumeBlockLists.size()];

    // 遍历perVolumeBlockLists
    for(Map.Entry<DatanodeStorage, BlockListAsLongs> kvPair : perVolumeBlockLists.entrySet()) {
      // 取出value：BlockListAsLongs
      BlockListAsLongs blockList = kvPair.getValue();
      // 将BlockListAsLongs封装成StorageBlockReport加入数据块汇报数组reports，
      // StorageBlockReport包含数据节点存储DatanodeStorage和其上数据块数组
      reports[i++] = new StorageBlockReport( kvPair.getKey(), blockList.getBlockListAsLongs());
      // 累加数据块数目totalBlockCount
      totalBlockCount += blockList.getNumberOfBlocks();
    }

    int numReportsSent = 0;
    int numRPCs = 0;
    boolean success = false;
    long brSendStartTime = now();
    long reportId = generateUniqueBlockReportId();
    try {
      // 根据数据块总数目判断是否需要多次发送消息
      if (totalBlockCount < dnConf.blockReportSplitThreshold) { // 如果数据块总数目在split阈值之下，则将所有的数据块汇报信息放在一个消息中发送
        // split阈值取参数dfs.blockreport.split.threshold，参数未配置的话默认为1000*1000

        // 通过NameNode代理bpNamenode的blockReport()方法向NameNode发送数据块汇报信息
        DatanodeCommand cmd = bpNamenode.blockReport( bpRegistration, bpos.getBlockPoolId(), reports, new BlockReportContext(1, 0, reportId));
        numRPCs = 1;
        // 发送的数据块汇报消息数numReportsSent设置为1
        numReportsSent = reports.length;
        // 将数据块汇报后返回的命令cmd加入到命令列表cmds
        if (cmd != null) {
          cmds.add(cmd);
        }
      } else {
        // 遍历reports，取出每个StorageBlockReport
        for (int r = 0; r < reports.length; r++) {
          StorageBlockReport singleReport[] = { reports[r] };
          // 通过NameNode代理bpNamenode的blockReport()方法向NameNode发送数据块汇报信息
          DatanodeCommand cmd = bpNamenode.blockReport( bpRegistration, bpos.getBlockPoolId(), singleReport, new BlockReportContext(reports.length, r, reportId));
          numReportsSent++;
          numRPCs++;
          // 将数据块汇报后返回的命令cmd加入到命令列表cmds
          if (cmd != null) {
            cmds.add(cmd);
          }
        }
      }
      success = true;
    } finally {
      // 计算数据块汇报耗时并记录在日志Log、数据节点Metrics指标体系中
      long brSendCost = now() - brSendStartTime;
      long brCreateCost = brSendStartTime - brCreateStartTime;
      dn.getMetrics().addBlockReport(brSendCost);
      final int nCmds = cmds.size();
      LOG.info((success ? "S" : "Uns") +
          "uccessfully sent block report 0x" +
          Long.toHexString(reportId) + ",  containing " + reports.length +
          " storage report(s), of which we sent " + numReportsSent + "." +
          " The reports had " + totalBlockCount + " total blocks and used " + numRPCs +
          " RPC(s). This took " + brCreateCost + " msec to generate and " + brSendCost +
          " msecs for RPC and NN processing. Got back " +
          ((nCmds == 0) ? "no commands" :  ((nCmds == 1) ? "one command: " + cmds.get(0) : (nCmds + " commands: " + Joiner.on("; ").join(cmds)))) + ".");
    }
    // 调用scheduleNextBlockReport()方法，调度下一次数据块汇报
    scheduleNextBlockReport(startTime);
    // 返回命令cmds
    return cmds.size() == 0 ? null : cmds;
  }

  private void scheduleNextBlockReport(long previousReportStartTime) {
    // If we have sent the first set of block reports, then wait a random
    // time before we start the periodic block reports.
    if (resetBlockReportTime) {
      lastBlockReport = previousReportStartTime -
          DFSUtil.getRandom().nextInt((int)(dnConf.blockReportInterval));
      resetBlockReportTime = false;
    } else {
      /* say the last block report was at 8:20:14. The current report
       * should have started around 9:20:14 (default 1 hour interval).
       * If current time is :
       *   1) normal like 9:20:18, next report should be at 10:20:14
       *   2) unexpected like 11:35:43, next report should be at 12:20:14
       */
      lastBlockReport += (now() - lastBlockReport) /
          dnConf.blockReportInterval * dnConf.blockReportInterval;
    }
  }

  DatanodeCommand cacheReport() throws IOException {
    // If caching is disabled, do not send a cache report
    if (dn.getFSDataset().getCacheCapacity() == 0) {
      return null;
    }
    // send cache report if timer has expired.
    DatanodeCommand cmd = null;
    final long startTime = Time.monotonicNow();
    if (startTime - lastCacheReport > dnConf.cacheReportInterval) {
      lastCacheReport = startTime;

      String bpid = bpos.getBlockPoolId();
      // 缓存块是基于 FSDatasetImpl管理的
      // 获取缓存块直接从 FSDatasetImpl里获取就行了
      List<Long> blockIds = dn.getFSDataset().getCacheReport(bpid); //
      long createTime = Time.monotonicNow();

      cmd = bpNamenode.cacheReport(bpRegistration, bpid, blockIds);
      long sendTime = Time.monotonicNow();
      long createCost = createTime - startTime;
      long sendCost = sendTime - createTime;
      dn.getMetrics().addCacheReport(sendCost);
      LOG.debug("CacheReport of " + blockIds.size()
          + " block(s) took " + createCost + " msec to generate and "
          + sendCost + " msecs for RPC and NN processing");
    }
    return cmd;
  }
  
  HeartbeatResponse sendHeartBeat() throws IOException {
    // 发送心跳的时候，带上了自己当前的存储信息
    StorageReport[] reports = dn.getFSDataset().getStorageReports(bpos.getBlockPoolId()); //

    // 核心代码
    return bpNamenode.sendHeartbeat(bpRegistration,
        reports,
        dn.getFSDataset().getCacheCapacity(), // 缓存可用的容量
        dn.getFSDataset().getCacheUsed(),// 已经使用了的缓存空间
        dn.getXmitsInProgress(),
        dn.getXceiverCount(),
        dn.getFSDataset().getNumFailedVolumes());
  }
  
  //This must be called only by BPOfferService
  void start() {
    // 之前在BlockPoolManager初始化的时候，startAll()方法已经被调用过了
    // 在这里DataNode在startDatanodeDaemon()方法再次调用了BlockPoolManager.startAll()方法，属于重复调用
    // 没什么用，也不会有什么问题，在这里会判断一下，如果这个BPServiceActor线程已经启动了，此时会直接返回，不会重复启动一个线程

    if ((bpThread != null) && (bpThread.isAlive())) {
      return;
    }
    bpThread = new Thread(this, formatThreadName());
    bpThread.setDaemon(true); // needed for JUnit testing
    bpThread.start();
    // 核心代码： BPServicector#run()
  }
  
  private String formatThreadName() {
    Collection<StorageLocation> dataDirs =
        DataNode.getStorageLocations(dn.getConf());
    return "DataNode: [" + dataDirs.toString() + "] " +
      " heartbeating to " + nnAddr;
  }
  
  //This must be called only by blockPoolManager.
  void stop() {
    shouldServiceRun = false;
    if (bpThread != null) {
        bpThread.interrupt();
    }
  }
  
  //This must be called only by blockPoolManager
  void join() {
    try {
      if (bpThread != null) {
        bpThread.join();
      }
    } catch (InterruptedException ie) { }
  }
  
  //Cleanup method to be called by current thread before exiting.
  private synchronized void cleanUp() {
    
    shouldServiceRun = false;
    IOUtils.cleanup(LOG, bpNamenode);
    bpos.shutdownActor(this);
  }

  private void handleRollingUpgradeStatus(HeartbeatResponse resp) throws IOException {
    RollingUpgradeStatus rollingUpgradeStatus = resp.getRollingUpdateStatus();
    if (rollingUpgradeStatus != null &&
        rollingUpgradeStatus.getBlockPoolId().compareTo(bpos.getBlockPoolId()) != 0) {
      // Can this ever occur?
      LOG.error("Invalid BlockPoolId " +
          rollingUpgradeStatus.getBlockPoolId() +
          " in HeartbeatResponse. Expected " +
          bpos.getBlockPoolId());
    } else {
      bpos.signalRollingUpgrade(rollingUpgradeStatus != null);
    }
  }

  /**
   * Main loop for each BP thread. Run until shutdown, forever calling remote NameNode functions.
   */
  private void offerService() throws Exception {

    while (shouldRun()) {
      try {
        final long startTime = now();

        // 每隔一段时间，发送心跳或阻塞报告
        // 当前时间减去上次发送心跳的时间 >= 心跳的间隔时间，就可以发送一次心跳了
        // 你配置了5秒钟发送一次心跳，在这里就会每隔5秒钟执行一次代码
        // 默认情况下，每隔3秒钟发送一次心跳
        if (startTime - lastHeartbeat >= dnConf.heartBeatInterval) {
          // All heartbeat messages include following info:
          // -- Datanode name
          // -- data transfer port
          // -- Total capacity
          // -- Bytes remaining
          lastHeartbeat = startTime; // 最近一次发送心跳的时间
          if (!dn.areHeartbeatsDisabledForTests()) {
            // 核心代码： 发送心跳过后，返回一个 HeartbeatResponse
            HeartbeatResponse resp = sendHeartBeat();
            assert resp != null;
            dn.getMetrics().addHeartbeat(now() - startTime);

            // 根据namenode返回的状态， active / standby
            // 更新当前BPServiceActive的状态，一组namenode是两个namenode,一个是active, 另一个是standby
            // 发送心跳的时候，如果感知到namenode的状态发生了切换，active -> standby
            // BPServiceActor 也是要更新自己的状态
            bpos.updateActorStatesFromHeartbeat( this, resp.getNameNodeHaState()); // TODO
            state = resp.getNameNodeHaState().getState();

            if (state == HAServiceState.ACTIVE) {
              handleRollingUpgradeStatus(resp);
            }

            long startProcessCommands = now();
            if (!processCommand(resp.getCommands())) { // TODO
              continue;
            }
            long endProcessCommands = now();
            if (endProcessCommands - startProcessCommands > 2000) {
              LOG.info("Took " + (endProcessCommands - startProcessCommands) + "ms to process " + resp.getCommands().length + " commands from NN");
            }
          }
        }

        // 上面代码是 心跳

        // ===============================数据块增量汇报============================================================

        // 数据块增量汇报 总结：
        // 数据块增量汇报是负责向NameNode发送心跳信息工作线程BPServiceActor中周期性的一个工作，
        // 它负责向NameNode及时汇报DataNode节点上数据块的变化情况，比如数据块正在接收、已接收或者已被删除。
        // 它的工作周期要小于正常的数据块汇报，目的就是为了能够让NameNode及时掌握DataNode上数据块变化情况，以便HDFS系统运行正常，略显机智！
        // 而且，当数据块增量汇报不成功时，下一个循环会接着立即发送数据块增量汇报，而不是等其下一个周期的到来，
        // 这显示了HDFS良好的容错性，是一个值得我们借鉴的设计方法。


        // 下面代码是不断的给namenode汇报自己的block，
        // 现在我们只是初探一下这块block report的源码，如果你要深入理解block report的源码，必须要完全看懂文件上传，
        // datanode是如何在上传数据的时候管理block的，然后才能看懂datanode是如何汇报block的
        // datanode汇报block的代码，基本上都是在BPServiceActor线程的offerService()方法内

        // 是否应立即汇报增量数据块信息的标志位sendImmediateIBR？
        // 当BPServiceActor工作线程创建时，这个标志位默认为false，即不会立即发送数据块增量汇报，而是周期性的到期才会发送。
        // 而当该发送数据块增量汇报时，无论标志位之前为true还是false，统一设置为false，因为此时数据块增量汇报已经发送了，下次没必要再立即发送了。
        // 而只有当数据块增量汇报不成功时，该标志位才会被设置为true，以便下次循环直接发送之前未成功的数据块增量汇报，而不用管数据块增量汇报的时间间隔是否到期。
        // 这个标志位就是为了在数据块增量汇报失败的情况下，下次循环中能立即发送出去，以便让NameNode及时了解DataNode数据块情况。

        // sendImmediateIBR是一个标志位，它标识着是否立即发送一个数据块增量汇报，在BPServiceActor工作线程初始化时默认为false
        // 如果标志位sendImmediateIBR为true，或者数据块增量汇报时间已到，
        // 数据块增量汇报时间间隔是心跳时间间隔的100倍，默认情况下是5分钟

        // 在block被datanode接收到了以后，不是同步进行通知namenode,而是先加入一个队列中，
        // 其实就是在BPServiceActor线程，默认每个5分钟将最近接收到，或者删除掉的block, 一次性批量上报给 namenode
        if (sendImmediateIBR || (startTime - lastDeletedReport > dnConf.deleteReportInterval)) {
          // 调用reportReceivedDeletedBlocks()方法发送数据块增量汇报
          reportReceivedDeletedBlocks(); // TODO
          // 设置上次数据块增量汇报时间lastDeletedReport为startTime
          lastDeletedReport = startTime;
        }

        // ===============================数据块汇报============================================================

        // 核心代码：向namenode汇报block report, 间隔时间6小时
        // 调用blockReport()方法，进行数据块汇报，放返回来自名字节点NameNode的相关命令cmds
        List<DatanodeCommand> cmds = blockReport(); // TODO
        // 调用processCommand()方法处理来自名字节点NameNode的相关命令cmds
        processCommand(cmds == null ? null : cmds.toArray(new DatanodeCommand[cmds.size()]));

        // 上报缓存block report , 间隔时间 10秒钟
        DatanodeCommand cmd = cacheReport();
        processCommand(new DatanodeCommand[]{ cmd });

        // 现在可以安全地开始扫描 block pool.
        // If it has already been started, this is a no-op.
        if (dn.blockScanner != null) {
          dn.blockScanner.addBlockPool(bpos.getBlockPoolId());
        }

        // 计算等待时间waitTime：心跳时间间隔减去上次心跳后截至到现在已过去的时间
        long waitTime = dnConf.heartBeatInterval -  (Time.now() - lastHeartbeat);
        synchronized(pendingIncrementalBRperStorage) {
          // 如果等待时间大于0，且不是立即发送数据块增量汇报
          if (waitTime > 0 && !sendImmediateIBR) {
            try {
              // 利用pendingIncrementalBRperStorage进行等待，并加synchronized关键字进行同步
              pendingIncrementalBRperStorage.wait(waitTime);
            } catch (InterruptedException ie) {
              LOG.warn("BPOfferService for " + this + " interrupted");
            }
          }
        } // synchronized
      } catch(RemoteException re) {
        String reClass = re.getClassName();
        if (UnregisteredNodeException.class.getName().equals(reClass) || DisallowedDatanodeException.class.getName().equals(reClass) || IncorrectVersionException.class.getName().equals(reClass)) {
          LOG.warn(this + " is shutting down", re);
          shouldServiceRun = false;
          return;
        }
        LOG.warn("RemoteException in offerService", re);
        try {
          long sleepTime = Math.min(1000, dnConf.heartBeatInterval);
          Thread.sleep(sleepTime);
        } catch (InterruptedException ie) {
          Thread.currentThread().interrupt();
        }
      } catch (IOException e) {
        LOG.warn("IOException in offerService", e);
      }
    } // while (shouldRun())
  } // offerService

  /**
   * 向相应的 NameNode 注册一个 bp （block Pool）
   *
   * bpDatanode需要在启动时向namenode注册，以便
   * 1) 告它现在提供的存储
   * 2) 接收一个 registrationID
   *  
   * 由namenode发出以识别已注册的datanode。
   * 
   * @throws IOException
   */
  void register() throws IOException {
    // The handshake() phase loaded the block pool storage
    // off disk - so update the bpRegistration object from that info

    // 调用了BPOfferServier.createRegistration()方法，创建了一个DatanodeRegistration对象
    // 大体相当于是一个datanode注册请求，这个里面包含了DataNodeID，这样的话，如果namenode收到了这个注册的请求(bpRegistration)
    // 那么就可以识别出来是哪个datanode发起的注册
    bpRegistration = bpos.createRegistration();

    LOG.info(this + " beginning handshake with NN");

    while (shouldRun()) {
      try {
        // Use returned registration from namenode with updated fields
        // 拿回来的 bpRegistration 对象里面肯定包含了一些namenode注册成功以后给datanode返回的一些东西
        bpRegistration = bpNamenode.registerDatanode(bpRegistration);
        break;
      } catch(EOFException e) {  // namenode might have just restarted
        LOG.info("Problem connecting to server: " + nnAddr + " :" + e.getLocalizedMessage());
        sleepAndLogInterrupts(1000, "connecting to server");
      } catch(SocketTimeoutException e) {
        LOG.info("Problem connecting to server: " + nnAddr);
        sleepAndLogInterrupts(1000, "connecting to server");
      }
    }
    
    LOG.info("Block pool " + this + " successfully registered with NN");
    // 标记 datanode注册成功
    bpos.registrationSucceeded(this, bpRegistration);

    // 随机短延迟 - 有助于分散所有DNs的block report
    // 延迟调度一次block report, 其实在这里会执行一次block report, 去汇报一次当前datanode节点上 全量的block有哪些
    // 这里会做一些时间的设定，在后面，就会在BPServiceActor.run()方法，线程的主流程里面，回去开始进行block report，第一次是全量汇报自己的block
    scheduleBlockReport(dnConf.initialBlockReportDelay);
  }


  private void sleepAndLogInterrupts(int millis,
      String stateString) {
    try {
      Thread.sleep(millis);
    } catch (InterruptedException ie) {
      LOG.info("BPOfferService " + this + " interrupted while " + stateString);
    }
  }

  /**
   * BPServiceActor线程刚启动时，
   *  1. 会在一个while循环内，完成连接NameNode并握手操作，即初始化。
   *      这里，是通过调用connectToNNAndHandshake()方法完成与NameNode的连接并进行两次握手的。
   *      值得一提的是，如果出现了IOException异常，会先将运行状态runningState设置为初始化失败INIT_FAILED，然后调用shouldRetryInit()方法判断初始化失败时是否可以重试：
   *          1、如果可以重试的话，记录error日志信息，线程休眠5s，并记录info日志信息，之后再进入循环重复执行之前的操作；
   *          2、不允许重试的情况下，将运行状态runningState设置为失败FAILED，退出循环，并返回。
   * 2. 接下来，设置运行状态runningState为正在运行RUNNING，进入另一个while循环，
   *       不停的调用offerService()方法，发送心跳给NameNode并接收来自NameNode的命令， 然后根据命令交给不同的组件去处理，
   *       循环的条件就是该线程的标志位shouldServiceRun为true，且dataNode的shouldRun()返回true；
   * 3. 而当循环过程中存在异常Exception的话，记录error日志，并休眠5s，然后继续循环，
   *       只有当shouldRun()方法返回false，才退出循环，
   *       设置运行状态runningState为已退出EXITED，
   *       最终调用cleanUp()方法释放占用的资源等。
   *
   *
   * 这个run()就是BPServiceActor线程的核心工作方法
   * 后面我们给大家去讲解这个datanode往namenode去注册，datanode定时的去给namenode发送心跳
   *
   * 这些逻辑，都是在这个BPServiceActor的run()方法里，它会去负责跟namenode进行通信
   */
  @Override
  public void run() {
    // 记录日志信息：starting to offer service
    LOG.info(this + " starting to offer service");

    // BPServiceActor是一个线程
    // 一旦对BPServiceActor这个线程执行了start()方法，启动了这个线程之后，
    // 实际上来说，这个线程的工作逻辑是在run()方法中

    try {
      // 在一个while循环内，完成连接NameNode并握手操作，即初始化
      while (true) {
        try {
          // 核心代码：连接到namenode以及进行握手，这个方法封装了整个datanode第1次启动进行注册的全过程
          connectToNNAndHandshake(); // TODO 核心代码
          // 如果注册的流程都结束了只有，那么这个方法就会执行结束，走一个break, 终结掉这个while死循环
          break;
        } catch (IOException ioe) {
          // 如果存在异常

          // 现将运行状态runningState设置为初始化失败INIT_FAILED
          runningState = RunningState.INIT_FAILED;

          // 调用shouldRetryInit()方法判断初始化失败时是否可以重试
          if (shouldRetryInit()) {
            // 记录error日志信息
            LOG.error("Initialization failed for " + this + " " + ioe.getLocalizedMessage());
            // 线程休眠5s，并记录info日志信息，之后再进入循环重复执行之前的操作
            sleepAndLogInterrupts(5000, "initializing");
          } else {
            // 不允许重试的情况下，将运行状态runningState设置为失败FAILED，退出循环，并返回
            runningState = RunningState.FAILED;
            LOG.fatal("Initialization failed for " + this + ". Exiting. ", ioe);
            return;
          }
        }
      }

      // 设置运行状态runningState为正在运行RUNNING
      runningState = RunningState.RUNNING;

      // 进入另一个while循环，不停的调用offerService()方法，
      // 发送心跳给NameNode并接收来自NameNode，然后根据命令交给不同的组件去处理
      // 循环的条件就是该线程的标志位shouldServiceRun为true，且dataNode的shouldRun()返回true

      // 这个线程其实会在这里进入一个while true的死循环，
      // BlockPoolOfferService,BlockPoolServiceActor
      // 这两个线程其实都是blockPool, 块管理相关的东西，还负责跟namenode之间的通信，
      // 面向对象的设计还是有一些缺陷的，因为跟namenode通信的线程设计成BlockPool相关的概念，不太合适，会让人觉得很迷惑
      while (shouldRun()) {
        try {
          // 核心代码： 每隔几秒种发送心跳， 每隔一段时间，汇报自己的block report
          offerService(); // TODO
        } catch (Exception ex) {
          // 存在异常的话，记录error日志，并休眠5s
          LOG.error("Exception in BPOfferService for " + this, ex);
          sleepAndLogInterrupts(5000, "offering service");
        }
      }
      // 设置运行状态runningState为已退出EXITED
      runningState = RunningState.EXITED;
    } catch (Throwable ex) {
      LOG.warn("Unexpected exception in block pool " + this, ex);
      runningState = RunningState.FAILED;
    } finally {
      LOG.warn("Ending block pool service for: " + this);
      // 清空，释放占用的资源
      cleanUp();
    }
  }

  private boolean shouldRetryInit() {
    return shouldRun() && bpos.shouldRetryInit();
  }

  private boolean shouldRun() {
    return shouldServiceRun && dn.shouldRun();
  }

  /**
   * Process an array of datanode commands
   * 
   * @param cmds an array of datanode commands
   * @return true if further processing may be required or false otherwise. 
   */
  boolean processCommand(DatanodeCommand[] cmds) {
    if (cmds != null) {
      for (DatanodeCommand cmd : cmds) {
        try {
          if (bpos.processCommandFromActor(cmd, this) == false) { //
            return false;
          }
        } catch (IOException ioe) {
          LOG.warn("Error processing datanode Command", ioe);
        }
      }
    }
    return true;
  }

  void trySendErrorReport(int errCode, String errMsg) {
    try {
      bpNamenode.errorReport(bpRegistration, errCode, errMsg);
    } catch(IOException e) {
      LOG.warn("Error reporting an error to NameNode " + nnAddr,
          e);
    }
  }

  /**
   * Report a bad block from another DN in this cluster.
   */
  void reportRemoteBadBlock(DatanodeInfo dnInfo, ExtendedBlock block)
      throws IOException {
    LocatedBlock lb = new LocatedBlock(block, 
                                    new DatanodeInfo[] {dnInfo});
    bpNamenode.reportBadBlocks(new LocatedBlock[] {lb});
  }

  void reRegister() throws IOException {
    if (shouldRun()) {
      // re-retrieve namespace info to make sure that, if the NN
      // was restarted, we still match its version (HDFS-2120)
      retrieveNamespaceInfo();
      // and re-register
      register();
      scheduleHeartbeat();
    }
  }

  private static class PerStoragePendingIncrementalBR {
    private final Map<Long, ReceivedDeletedBlockInfo> pendingIncrementalBR =
        Maps.newHashMap();

    /**
     * Return the number of blocks on this storage that have pending
     * incremental block reports.
     * @return
     */
    int getBlockInfoCount() {
      return pendingIncrementalBR.size();
    }

    /**
     * Dequeue and return all pending incremental block report state.
     * @return
     */
    ReceivedDeletedBlockInfo[] dequeueBlockInfos() {
      ReceivedDeletedBlockInfo[] blockInfos =
          pendingIncrementalBR.values().toArray(
              new ReceivedDeletedBlockInfo[getBlockInfoCount()]);

      pendingIncrementalBR.clear();
      return blockInfos;
    }

    /**
     * Add blocks from blockArray to pendingIncrementalBR, unless the
     * block already exists in pendingIncrementalBR.
     * @param blockArray list of blocks to add.
     * @return the number of missing blocks that we added.
     */
    int putMissingBlockInfos(ReceivedDeletedBlockInfo[] blockArray) {
      int blocksPut = 0;
      for (ReceivedDeletedBlockInfo rdbi : blockArray) {
        if (!pendingIncrementalBR.containsKey(rdbi.getBlock().getBlockId())) {
          pendingIncrementalBR.put(rdbi.getBlock().getBlockId(), rdbi);
          ++blocksPut;
        }
      }
      return blocksPut;
    }

    /**
     * Add pending incremental block report for a single block.
     * @param blockInfo
     */
    void putBlockInfo(ReceivedDeletedBlockInfo blockInfo) {
      pendingIncrementalBR.put(blockInfo.getBlock().getBlockId(), blockInfo);
    }

    /**
     * Remove pending incremental block report for a single block if it
     * exists.
     *
     * @param blockInfo
     * @return true if a report was removed, false if no report existed for
     *         the given block.
     */
    boolean removeBlockInfo(ReceivedDeletedBlockInfo blockInfo) {
      return (pendingIncrementalBR.remove(blockInfo.getBlock().getBlockId()) != null);
    }
  }

  void triggerBlockReport(BlockReportOptions options) throws IOException {
    if (options.isIncremental()) {
      LOG.info(bpos.toString() + ": scheduling an incremental block report.");
      synchronized(pendingIncrementalBRperStorage) {
        sendImmediateIBR = true;
        pendingIncrementalBRperStorage.notifyAll();
      }
    } else {
      LOG.info(bpos.toString() + ": scheduling a full block report.");
      synchronized(pendingIncrementalBRperStorage) {
        lastBlockReport = 0;
        pendingIncrementalBRperStorage.notifyAll();
      }
    }
  }
}
